-- thread.lua
-- Created by KingMars Jan/31/2011 
-- manage the coroutine thread 
module("THREAD_MGR", package.seeall)

-- 线程池大小
local THREAD_POOL_SIZE = 40;

--[[
{
    status = 'run' | 'suspend' | 'sleep' ;
    param = par_list
    sleep_time
    coroutine
}
--]]
local m_all_named_threads = {}

--[[
   {
        exc_count,   -- 剩余执行次数
        time,        -- 距离下一次执行的时间
        func,        -- timer回调
        params,      -- 参数
   }
--]]
local m_all_timers = {}

-- 当前执行的线程
local m_current_thread = nil;

-- 线程池，预先创建好的一些线程
local m_thread_pool = {}

-- globla sleep function
function sleep(time)
    if m_current_thread == nil then
        logger:warn('Can not sleep in main thread');
    else
       if time == 0 then
           coroutine.yield();
       else
           m_current_thread.status = 'sleep';
           m_current_thread.sleep_time = time;
           coroutine.yield();
       end
    end
end

_G.sleep = sleep;

-- do when update this source file

function _get_module_name()
    return "manager.thread_mgr"
end

-- show the information of thread manager
function dump()
    print('---------------- THREAD DUMP -----------------');
    print('All named thread num:', table.size(m_all_named_threads));
    print('All timer num:', table.size(m_all_timers));
    print('All thread pool:', table.size(m_thread_pool));
    print('----------------------------------------------');
end

-- create thread
function create_thread(fun, ...) 
    return create_named_thread(nil, fun, ...);
end

-- create named thread (new)
function create_named_thread(name, fun, ...) 
    name = name or UTIL.gen_unique_string('thread');
    if m_all_named_threads[name] ~= nil then
        logger:error('A thread named '..name..' already existed.');
        return nil;
    end

    local co = coroutine.create(fun);
    
    local thread = 
    {
        param = {...};
        status = 'run';
        name = name,
        coroutine = co;
    }

    m_all_named_threads[name] = thread;
    return thread;
end

-- 发布一个任务，放入线程池中
function dispatch_thread(name, fun, ...)

    name = name or UTIL.gen_unique_string('service');
    if not is_function(fun) then
        error('fun must be a function');
    end

    -- 获取一个空闲的线程
    for _, v in ipairs(m_thread_pool) do
        if v.idle then
            v.service.fun = fun;
            v.service.params = {...};
            v.idle = false;
            v.status = 'run';
            return v;
        end
    end

    logger:warn('Has no idle thread in pool.');

    -- 直接创建
    return create_named_thread(name, fun, ...);
end

-- 暂停timer
function pause_timer(timer)
    timer.pause = true;
end

-- 回复timer
function resume_timer(timer)
    timer.pause = nil;
end

-- 暂停线程
function pause_thread(thread)
    if is_string(thread) then
        thread = m_all_named_threads[thread];
    end

    if get_m_current_thread() == thread then
        logger:warn('Can not pause current running thread.');
        return;
    end

    if thread ~= nil then
        if thread.status == 'pause' then
            logger:warn('Thread '..thread.name..' has already been paused');
            return;
        end

        thread.pause_status = thread.status;
        thread.status = 'pause';
    end
end

-- 恢复线程
function resume_thread(thread)
    if is_string(thread) then
        thread = m_all_named_threads[thread];
    end

    if thread ~= nil then
        if thread.status ~= 'pause' then
            logger:warn('Thread can not be resumed when it is not paused');
        else
            thread.status = thread.pause_status;
            thread.pause_status = nil;
        end
    end
end

-- find a thread through name
function get_thread(name)
    return m_all_named_threads[name];
end

local function timer_func(params)
    local func = params.func;
    if not params.call_immediatly then
        sleep(params.duration);
    end 

    for i = 1, params.n do
        func(params.func_param);
        sleep(params.duration);
    end
end

local function once_timer_func(func, duration, ...)
    sleep(duration);
    func(...);
end

-- create a timer call the [func] [n] times per [duration]  second 
function create_timer(func, duration, n, ...)
    local timer = {
        func = func,
        count = n,
        time = duration,
        duration = duration,
        params = {...},
    }

    table.insert(m_all_timers, timer);
    return timer;
end

-- create a thread timer
function create_once_thread_timer(func, duration, ...)
    THREAD_MGR.dispatch_thread(nil, once_timer_func, func, duration, ...);
end

-- create a once timer
function create_once_timer(func, duration, ...)
    local timer = {
        func = func,
        count = 1,
        time = duration,
        duration = duration,
        params = {...},
    }

    table.insert(m_all_timers, timer);
    return timer;
end

-- destroy a timer
function destroy_timer(timer)
    array.remove(m_all_timers, timer);
end

-- destroy thread
function destroy_thread(thread)
    if is_table(thread) then
        thread = thread.name;
    end
    assert(is_string(thread));
    m_all_named_threads[thread] = nil; 

    -- if destroy current thread, yield now
    if m_current_thread == thread then
        coroutine.yield();
    end
end

-- get current thread
function get_m_current_thread()
    return m_current_thread;
end

-- run the thread one time
local function run_thread(co, param)
    local success, errmsg = coroutine.resume(co, param);
    if not success then
        colorPrint(debug.traceback(co, errmsg, 1), PrintColor.IRED);
    end
end

-- run the thread with varargs
local function run_thread_ex(co, param)

    local function resume_func(...)
        local success, errmsg = coroutine.resume(co, ...);
        if not success then
            colorPrint(debug.traceback(co, errmsg, 1), PrintColor.IRED);
        end
    end

    resume_func(unpack(param));
end

-- update all thre threads
function update(time)

    for k, v in pairs(m_all_named_threads) do
        local co = v.coroutine;
        if coroutine.status(co) == 'dead' then
            destroy_thread(k);

        elseif v.status == 'run' then
            m_current_thread = v;
            run_thread_ex(co, v.param);

        elseif v.status == 'sleep' then
            local sleep_time = v.sleep_time;
            sleep_time = sleep_time - time;

             -- is sleep over?
            if sleep_time < 0 then
                v.status = 'run';
                v.sleep_time = 0;
                m_current_thread = v;
                run_thread_ex(co, v.param);
            else
                v.sleep_time = sleep_time;
            end
        end
    end

    for _, v in ipairs(m_thread_pool) do
        if not v.idle then
            local co = v.coroutine;

            if v.status == 'run' then
                m_current_thread = v;
                run_thread_ex(co, {v.service});
            elseif v.status == 'sleep' then
                local sleep_time = v.sleep_time;
                sleep_time = sleep_time - time;

                if sleep_time < 0 then
                    v.status = 'run';
                    v.sleep_time = 0;
                    m_current_thread = v;
                    run_thread_ex(co, v.service);
                else
                    v.sleep_time = sleep_time;
                end
            end
        end
    end

    m_current_thread = nil;

    -- 更新所有timer
    local del_timers = nil;
    for _, v in ipairs(m_all_timers) do
        if not v.pause then
            if v.time < 0 then
                -- 需要执行这个timer
                v.func(unpack(v.params));
                v.time = v.time + v.duration;
                if v.count then
                    v.count = v.count - 1;
                    if v.count <= 0 then
                        del_timers = del_timers or {}
                        table.insert(del_timers, v);
                    end
                end
            end
            v.time = v.time - time;
        end
    end

    if del_timers ~= nil then
        for _, v in ipairs(del_timers) do
            array.remove(m_all_timers, v);
        end
    end

    -- set current thread nil
    m_current_thread = nil;
end

-- 线程池中的服务线程
local function _service_thread(service)
    while true do
        service.fun(unpack(service.params));
        service.fun = nil;
        service.params = nil;
        m_current_thread.idle = true;
        sleep(0);
    end
end

-- 初始化线程池
local function _init_thread_pool()
    for i = 1, THREAD_POOL_SIZE do
        local co = coroutine.create(_service_thread);

        local thread = 
        {
            service = {},
            coroutine = co,
            idle = true,
        }

        table.insert(m_thread_pool, thread);
    end
end

-- 获取当前线程
function _G.get_current_thread()
    return m_current_thread;
end

-- 全局判断是否在线程中的方法
function _G.is_in_thread()
    return m_current_thread ~= nil;
end

_init_thread_pool();

